package org.ewancle.springboothivehdfs.service;

import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.io.IOUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;

import java.io.*;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Slf4j
@Service
public class HDFSService {

    private final FileSystem fileSystem;
    private final ExecutorService executorService;

    @Autowired
    public HDFSService(FileSystem fileSystem) {
        this.fileSystem = fileSystem;
        // 创建线程池优化并发操作
        this.executorService = Executors.newFixedThreadPool(10);
    }

    /**
     * 创建目录
     * @param path 路径
     * @return 是否成功
     */
    public boolean mkdir(String path) {
        try {
            Path hdfsPath = new Path(path);
            if (!fileSystem.exists(hdfsPath)) {
                return fileSystem.mkdirs(hdfsPath);
            }
            return true;
        } catch (Exception e) {
            log.error("Create directory failed: {}", path, e);
            return false;
        }
    }

    /**
     * 创建带权限的目录
     * @param path 路径
     * @param permission 权限字符串，例如 "755"
     * @return 是否成功
     */
    public boolean mkdirWithPermission(String path, String permission) {
        try {
            Path hdfsPath = new Path(path);
            if (!fileSystem.exists(hdfsPath)) {
                FsPermission fsPermission = new FsPermission(permission);
                return fileSystem.mkdirs(hdfsPath, fsPermission);
            }
            return true;
        } catch (Exception e) {
            log.error("Create directory with permission failed: {}", path, e);
            return false;
        }
    }

    /**
     * 上传文件
     * @param srcFile 源文件
     * @param destPath 目标路径
     * @return 是否成功
     */
    public boolean uploadFile(File srcFile, String destPath) {
        InputStream inputStream = null;
        try {
            inputStream = new BufferedInputStream(new FileInputStream(srcFile));
            return uploadFile(inputStream, destPath);
        } catch (Exception e) {
            log.error("Upload file failed: {} -> {}", srcFile.getAbsolutePath(), destPath, e);
            return false;
        } finally {
            IOUtils.closeStream(inputStream);
        }
    }

    /**
     * 通过输入流上传文件
     * @param inputStream 输入流
     * @param destPath 目标路径
     * @return 是否成功
     */
    public boolean uploadFile(InputStream inputStream, String destPath) {
        try {
            Path hdfsPath = new Path(destPath);
            // 确保目录存在
            Path parent = hdfsPath.getParent();
            if (parent != null && !fileSystem.exists(parent)) {
                fileSystem.mkdirs(parent);
            }

            // 使用缓冲写入提高性能
            try (OutputStream outputStream = new BufferedOutputStream(fileSystem.create(hdfsPath, true))) {
                IOUtils.copyBytes(inputStream, outputStream, 4096, false);
            }
            return true;
        } catch (Exception e) {
            log.error("Upload file by InputStream failed: {}", destPath, e);
            return false;
        }
    }

    /**
     * 上传Web多部分文件
     * @param file 多部分文件
     * @param destPath 目标路径
     * @return 是否成功
     */
    public boolean uploadFile(MultipartFile file, String destPath) {
        try {
            return uploadFile(file.getInputStream(), destPath);
        } catch (Exception e) {
            log.error("Upload MultipartFile failed: {} -> {}", file.getOriginalFilename(), destPath, e);
            return false;
        }
    }

    /**
     * 异步上传文件
     * @param file 多部分文件
     * @param destPath 目标路径
     * @return CompletableFuture对象
     */
    public CompletableFuture<Boolean> uploadFileAsync(MultipartFile file, String destPath) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                return uploadFile(file.getInputStream(), destPath);
            } catch (Exception e) {
                log.error("Async upload file failed: {} -> {}", file.getOriginalFilename(), destPath, e);
                return false;
            }
        }, executorService);
    }

    /**
     * 下载文件
     * @param srcPath HDFS路径
     * @param destFile 本地文件
     * @return 是否成功
     */
    public boolean downloadFile(String srcPath, File destFile) {
        OutputStream outputStream = null;
        try {
            Path hdfsPath = new Path(srcPath);
            if (!fileSystem.exists(hdfsPath)) {
                return false;
            }

            // 确保目标目录存在
            File parentDir = destFile.getParentFile();
            if (parentDir != null && !parentDir.exists()) {
                parentDir.mkdirs();
            }

            outputStream = new BufferedOutputStream(new FileOutputStream(destFile));
            try (InputStream inputStream = fileSystem.open(hdfsPath)) {
                IOUtils.copyBytes(inputStream, outputStream, 4096, false);
            }
            return true;
        } catch (Exception e) {
            log.error("Download file failed: {} -> {}", srcPath, destFile.getAbsolutePath(), e);
            return false;
        } finally {
            IOUtils.closeStream(outputStream);
        }
    }

    /**
     * 读取文件内容为字符串
     * @param filePath HDFS文件路径
     * @return 文件内容
     */
    public String readFile(String filePath) throws IOException {
        Path hdfsPath = new Path(filePath);
        if (!fileSystem.exists(hdfsPath)) {
            throw new FileNotFoundException("File not found: " + filePath);
        }

        StringBuilder content = new StringBuilder();
        try (BufferedReader reader = new BufferedReader(new InputStreamReader(fileSystem.open(hdfsPath)))) {
            String line;
            while ((line = reader.readLine()) != null) {
                content.append(line).append("\n");
            }
        }
        return content.toString();
    }

    /**
     * 删除文件或目录
     * @param path 路径
     * @param recursive 是否递归删除
     * @return 是否成功
     */
    public boolean delete(String path, boolean recursive) {
        try {
            Path hdfsPath = new Path(path);
            if (fileSystem.exists(hdfsPath)) {
                return fileSystem.delete(hdfsPath, recursive);
            }
            return false;
        } catch (Exception e) {
            log.error("Delete failed: {}", path, e);
            return false;
        }
    }

    /**
     * 获取目录列表
     * @param path 路径
     * @return 文件状态列表
     */
    public List<FileStatus> listFiles(String path) throws IOException {
        Path hdfsPath = new Path(path);
        List<FileStatus> result = new ArrayList<>();

        if (!fileSystem.exists(hdfsPath)) {
            return result;
        }

        FileStatus[] statuses = fileSystem.listStatus(hdfsPath);
        if (statuses != null) {
            for (FileStatus status : statuses) {
                result.add(status);
            }
        }

        return result;
    }

    /**
     * 检查文件是否存在
     * @param path 路径
     * @return 是否存在
     */
    public boolean exists(String path) {
        try {
            return fileSystem.exists(new Path(path));
        } catch (Exception e) {
            log.error("Check file exists failed: {}", path, e);
            return false;
        }
    }

    /**
     * 重命名文件或目录
     * @param srcPath 源路径
     * @param destPath 目标路径
     * @return 是否成功
     */
    public boolean rename(String srcPath, String destPath) {
        try {
            Path src = new Path(srcPath);
            Path dest = new Path(destPath);
            return fileSystem.rename(src, dest);
        } catch (Exception e) {
            log.error("Rename failed: {} -> {}", srcPath, destPath, e);
            return false;
        }
    }

    /**
     * 获取文件状态
     * @param path 路径
     * @return 文件状态
     */
    public FileStatus getFileStatus(String path) throws IOException {
        Path hdfsPath = new Path(path);
        return fileSystem.getFileStatus(hdfsPath);
    }

    /**
     * 关闭服务
     */
    public void close() {
        if (executorService != null) {
            executorService.shutdown();
        }

        if (fileSystem != null) {
            try {
                fileSystem.close();
            } catch (IOException e) {
                log.error("Close FileSystem failed", e);
            }
        }
    }
}
